package com.dmp.storm.bolt.db;

import com.dmp.storm.bolt.BaseMapBolt;
import com.dmp.util.DateUtils;
import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Element;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.*;

/** 
 * @ClassName InsertBatchDBBolt
 * @Description 批量插入普通方式
 *
 * @author  sunzy 
 * @date 2016年4月13日 下午4:54:21 
 * @version 1.0 
 *
 */
@SuppressWarnings("serial")
public class DBCreateTableBolt extends BaseMapBolt {
	private static final Logger logger = LoggerFactory.getLogger(DBCreateTableBolt.class);

	private Connection connection;
	//数据库配置
	private String driverClaz = "com.mysql.jdbc.Driver";
    private String url;
    private String user;
    private String pwd;
    
	private String sql;
	private String tableNamePrefix = "";
	private String dateFieldName = "";
	//数据过滤配置
	private String filterFieldName = "";
	private String filterFieldValue = "";
	//缓存
	private static CacheManager cacheManager;  
	private static String cacheName = "hasTableCache";
	private static Cache hasTableCache;

	private final static String hasTableSql = "select count(1) from ? where 1 != 1";
	private List<String> filterList = new ArrayList<String>();
	
	@Override
	public void prepare() {
		try {
        	Class.forName(getDriverClaz());
            connection = DriverManager.getConnection(url, user, pwd);
            
            cacheManager = CacheManager.create();
            cacheManager.addCache(cacheName);
            hasTableCache= cacheManager.getCache(cacheName);  
            
            //
            if(StringUtils.isNotBlank(getFilterFieldName()) && StringUtils.isNotBlank(getFilterFieldValue())){
            	String[] filters = StringUtils.split(getFilterFieldValue(), ",");
            	filterList = Arrays.asList(filters);
            }
        } catch (Exception e) {
        	logger.error(e.getMessage());
        	
            System.exit(-1);
        }
	}
	
	/**
	 * @author sunzy
	 * @date 2016年2月24日 下午2:04:51
	 * @desc 此处不需要处理，空运行
	 *
	 * @return
	 * 
	 * @see com.dmp.storm.base.BaseStringSpout#process()
	 */
	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		if(StringUtils.isBlank(getFilterFieldName()) || 
				(StringUtils.isNotBlank(getFilterFieldName()) 
						&& StringUtils.isNotBlank(getFilterFieldValue()) && filterList.contains("" + map.get(getFilterFieldName())))){
			
			createTable(map);
		}
			
		return map;
	}
	
	/**
	 * 配置了日期字段，我们就动态建表；如果没有配置此字段，就不建表
	 * 
	 * @param map
	 * @return
	 */
	private boolean createTable(Map<String, Object> map){
		boolean isSuccess = true;
		String tableName = buildTableName(map);
		Statement hasTableStmt = null;
		try{
			if(StringUtils.isNotBlank(tableName)){
				if(!get(tableName)){
					String hasTableSqlStr = StringUtils.replace(hasTableSql, "?", tableName);
					hasTableStmt = connection.createStatement();
					hasTableStmt.executeQuery(hasTableSqlStr);
					put(tableName, Boolean.TRUE.toString());
				}
			}
		}catch(Exception e){
			//表不存在
			Statement createTableStmt = null;
			try{
				String createTableSql = StringUtils.replace(getSql(), "?", tableName);
				createTableStmt = connection.createStatement();
				createTableStmt.execute(createTableSql);
				put(tableName, Boolean.TRUE.toString());
			}catch(Exception ex){
				logger.error(ex.getMessage());
				isSuccess = false;
			} finally{
				try{
					if(createTableStmt != null){
						createTableStmt.close();
					}
				}catch(Exception ex){
				}
			}
		} finally {
			try{
				if(hasTableStmt != null){
					hasTableStmt.close();
				}
			}catch(Exception e){
			}
		}
		
		return isSuccess;
	}
	
	private String buildTableName(Map<String, Object> map){
		String tableName = "";
		try{
			if(StringUtils.isNotBlank(getDateFieldName())){
				Object dateValue = map.get(getDateFieldName());
				//约定日期字段为字符类型
				if(dateValue != null){
					String dateStr = StringUtils.substring(dateValue.toString(), 0, StringUtils.indexOf(dateValue.toString(), "."));
					Date recordDate = DateUtils.parseDate(dateStr, "yyyy-MM-dd HH:mm:ss");
					tableName = getTableNamePrefix() + DateUtils.formatDate(recordDate, "yyyyMMdd");
				}
			}
		} catch(Exception e){
			logger.error(e.getMessage());
		}
		
		return tableName;
	}
	
	private void put(String key, String value) {  
		hasTableCache.put(new Element(key, value));  
	}

	private Boolean get(String key) {
		Element elem = hasTableCache.get(key);
		
	    return elem != null;  
	}
	
	public String getDriverClaz() {
		return driverClaz;
	}

	public void setDriverClaz(String driverClaz) {
		this.driverClaz = driverClaz;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public String getUser() {
		return user;
	}

	public void setUser(String user) {
		this.user = user;
	}

	public String getPwd() {
		return pwd;
	}

	public void setPwd(String pwd) {
		this.pwd = pwd;
	}

	public String getSql() {
		return sql;
	}

	public void setSql(String sql) {
		this.sql = sql;
	}

	public String getTableNamePrefix() {
		return tableNamePrefix;
	}

	public void setTableNamePrefix(String tableNamePrefix) {
		this.tableNamePrefix = tableNamePrefix;
	}

	public String getDateFieldName() {
		return dateFieldName;
	}

	public void setDateFieldName(String dateFieldName) {
		this.dateFieldName = dateFieldName;
	}

	public String getFilterFieldName() {
		return filterFieldName;
	}

	public void setFilterFieldName(String filterFieldName) {
		this.filterFieldName = filterFieldName;
	}

	public String getFilterFieldValue() {
		return filterFieldValue;
	}

	public void setFilterFieldValue(String filterFieldValue) {
		this.filterFieldValue = filterFieldValue;
	}

}
